package com.wmh.baseservice.common.redis.mq.core.job;


import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.wmh.baseservice.common.redis.mq.core.RedisMqTemplate;
import com.wmh.baseservice.common.redis.mq.core.stream.RedisStreamFailHandlerFactory;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;

import java.util.List;

/**
 * redis 未 ack 消息重发定时任务
 *
 * @author 审心
 */
@Slf4j
@AllArgsConstructor
public class RedisPendingMessageResendJob {
    private static final String LOCK_KEY = "redis:pending:msg:lock";

    private final RedissonClient redissonClient;

    private final RedisMqTemplate redisMqTemplate;

    private final RedisStreamFailHandlerFactory redisStreamFailHandlerFactory;

    /**
     * 一分钟执行一次,这里选择每分钟的35秒执行，是为了避免整点任务过多的问题
     */
    @Scheduled(cron = "35 * * * * ?")
    public void messageResend() {
        RLock lock = redissonClient.getLock(LOCK_KEY);
        // 尝试加锁
        if (lock.tryLock()) {
            try {
                execute();
            } catch (Exception ex) {
                log.error("[RedisPendingMessageResendJob][执行异常]", ex);
            } finally {
                lock.unlock();
            }
        }
    }

    /**
     * 执行重发逻辑
     */
    private void execute() {
        StringRedisTemplate template = redisMqTemplate.getStringRedisTemplate();
        StreamOperations<String, Object, Object> ops = template.opsForStream();
        for (String key : RedisMqTemplate.STREAM_KEY_LIST) {
            if (Boolean.FALSE.equals(template.hasKey(key))) {
                continue;
            }
            StreamInfo.XInfoGroups groups = ops.groups(key);
            groups.forEach(group -> {
                PendingMessages pendingMessages = ops.pending(key, group.groupName(), Range.unbounded(), group.pendingCount());
                for (PendingMessage pendingMessage : pendingMessages) {
                    JSONObject json = getMessage(ops, key, pendingMessage);
                    // 最大消费时间
                    Long maxConsumptionTime = json.getLong("maxConsumptionTime");
                    if (pendingMessage.getElapsedTimeSinceLastDelivery().getSeconds() < maxConsumptionTime) {
                        continue;
                    }
                    // 最大重试次数
                    Long maxRetryCount = json.getLong("maxRetryCount");
                    // 当前重试次数
                    Long retryCount = json.getLong("retryCount");
                    // 超过了重试次数不再重新投递
                    if (++retryCount > maxRetryCount) {
                        log.warn("[RedisPendingMessageResendJob] 重试失败 [消息编号 : {} ] [消费者 : {}] [消息体 : {} ]"
                                , pendingMessage.getIdAsString(), pendingMessage.getConsumer(), json);
                        // 消息失败处理
                        redisStreamFailHandlerFactory.handler(key, json);
                    } else {
                        // 重设重试次数
                        json.put("retryCount", retryCount);
                        // 重新投递消息
                        redisMqTemplate.streamTrim(key, json.getLong("maxSize"));
                        RecordId id = ops.add(StreamRecords.newRecord()
                                .ofObject(json.toJSONString())
                                .withStreamKey(key));
                        log.info("[RedisPendingMessageResendJob] 重新投递成功 [旧编号 : {} ] [新编号 : {} ]  [消费者 : {}] [消息体 : {} ]"
                                , pendingMessage.getIdAsString(), id, pendingMessage.getConsumer(), json);
                    }
                    // ack 旧信息
                    ops.acknowledge(key, group.groupName(), pendingMessage.getIdAsString());
                }
            });
        }
    }

    private JSONObject getMessage(StreamOperations<String, Object, Object> ops, String key, PendingMessage pendingMessage) {
        Range.Bound<String> idBound = Range.Bound.inclusive(pendingMessage.getIdAsString());
        List<ObjectRecord<String, String>> messages = ops.range(String.class, key, Range.of(idBound, idBound));
        // 解析消息体未 json
        return JSON.parseObject(messages.get(0).getValue());
    }
}
